package cn.money.youtube

import org.apache.parquet.format.LogicalTypes.TimeUnits
import org.apache.spark.{SparkConf, SparkContext}

import java.io.StringReader
import java.util.concurrent.TimeUnit

/**
 * Spark RDD处理示例
 */
object RddAnalysis {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("RDD Process").setMaster("local[*]")
        val sc = new SparkContext(conf)

        // 读取本地文件创建RDD
        val youtubeVideosRDD = {
            sc.textFile("""E:\05.git_project\dataset\youtube""")
        }

        // 统计不同分类Youtube视频的喜欢人数、不喜欢人数
        // 1. 添加行号
        // 创建计数器
        val rownumAcc = sc.longAccumulator("rownum")
        // 带上行号
        youtubeVideosRDD.map(line => {
                rownumAcc.add(1)
                rownumAcc.value -> line
            })
            // 过滤掉第一行
            .filter(_._1 != 1)
            // 去除行号
            .map(_._2)
            // 过滤掉非法的数据
            .filter(line => {
                val fields = line.split("\001")
                val try1 = scala.util.Try(fields(8).toLong)
                val try2 = scala.util.Try(fields(9).toLong)

                if(try1.isFailure || try2.isFailure)
                    false
                else
                    true
            })
            // 读取三个字段（视频分类、喜欢的人数、不喜欢的人数
            .map(line => {
                // 按照\001解析CSV
                val fields = line.split("\001")
                // 取第4个（分类）、第8个（喜欢人数）、第9个（不喜欢人数）
                // (分类id, 喜欢人数, 不喜欢人数)
                (fields(4), fields(8).toLong, fields(9).toLong)
            })
            // 按照分类id分组
            .groupBy(_._1)
            .map(t => {
                val result = t._2.reduce((r1, r2) => {
                    (r1._1, r1._2 + r2._2, r1._3 + r2._3)
                })
                result
            })
            .foreach(println)

        TimeUnit.HOURS.sleep(1)
    }
}
